package xin.marcher.module.message.service.impl;

import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;
import xin.marcher.module.common.cache.MessageCache;
import xin.marcher.module.common.constants.CacheConstant;
import xin.marcher.module.common.constants.CollectionSize;
import xin.marcher.module.common.constants.CoreConstant;
import xin.marcher.module.common.constants.RocketMqConstant;
import xin.marcher.module.common.domain.BinlogData;
import xin.marcher.module.common.domain.TraceData;
import xin.marcher.module.common.enums.construction.ProductExceptionCode;
import xin.marcher.module.common.enums.message.BinlogType;
import xin.marcher.module.common.enums.message.MessageTypeEnum;
import xin.marcher.module.common.exception.BaseBizException;
import xin.marcher.module.common.message.DataChangeMessage;
import xin.marcher.module.common.redis.RedisCache;
import xin.marcher.module.common.utils.DateFormatUtil;
import xin.marcher.module.common.utils.SnowflakeIdWorker;
import xin.marcher.module.message.cache.CacheQueue;
import xin.marcher.module.message.converter.DataMessageConverter;
import xin.marcher.module.message.domain.bo.DataChangeMessageConfigBO;
import xin.marcher.module.message.domain.bo.DataMessageBO;
import xin.marcher.module.message.domain.bo.DataSendMessageBO;
import xin.marcher.module.message.domain.entity.*;
import xin.marcher.module.message.mq.producer.DataMessageProducer;
import xin.marcher.module.message.repository.DataChangeRepository;
import xin.marcher.module.message.repository.DataRefreshRepository;
import xin.marcher.module.message.service.MessageService;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.BlockingQueue;
import java.util.stream.Collectors;

/**
 * 消息业务实现类
 */
@Service
public class MessageServiceImpl implements MessageService {

    @Resource
    private DataChangeRepository dataChangeRepository;

    @Resource
    private DataMessageConverter dataMessageConverter;

    @Autowired
    private DataMessageProducer dataMessageProducer;

    @Autowired
    private RedisCache redisCache;

    @Autowired
    private DataRefreshRepository dataRefreshRepository;


    @Autowired
    private CacheQueue cacheQueue;

    /**
     * 存储一条待发送的消息信息(待发送完成后删除)
     *
     * @param dataMessageBO
     */
    @Override
    public void saveMessage(DataMessageBO dataMessageBO) {
        dataChangeRepository.saveDataMessageDetail(dataMessageBO);
    }

    /**
     * 删除一条已发送的消息信息
     *
     * @param dataMessageBO
     */
    @Override
    public void deleteMessage(DataMessageBO dataMessageBO) {
        dataChangeRepository.deleteDataMessageDetail(dataMessageBO.getMessageNo());
    }

    /**
     * 获取对应的外部消息对象信息
     *
     * @param messageNo
     * @return
     */
    @Override
    public DataMessageBO getDataMessageDetail(String messageNo) {
        DataMessageDetailDO dataMessageDetail = dataChangeRepository.getDataMessageDetail(messageNo);
        return dataMessageConverter.converterBO(dataMessageDetail);
    }

    /**
     * 获取需要发送的消息对象
     *
     * @param messageBO
     */
    @Override
    public List<DataSendMessageBO> getSendDataMessage(DataMessageBO messageBO) {
        List<DataSendMessageBO> dataChangeMessageConfig = dataChangeRepository.getDataChangeMessageConfig(messageBO);
        return dataChangeMessageConfig;
    }


    /**
     * 保存消息详细信息
     *
     * @param dataChangeMessages
     * @param binlogData
     */
    public void saveDataMessageDetail(List<DataChangeMessage> dataChangeMessages, BinlogData binlogData) {

        List<DataMessageBO> dataMessageBOS = converterDataMessageBOList(dataChangeMessages, binlogData);

        dataChangeRepository.saveDataMessageDetail(dataMessageBOS);

    }

    /**
     * 转换消息详细信息
     *
     * @param dataChangeMessages
     * @param binlogData
     * @return
     */
    private List<DataMessageBO> converterDataMessageBOList(List<DataChangeMessage> dataChangeMessages, BinlogData binlogData) {
        List<DataMessageBO> dataMessageBOS = new ArrayList<>(dataChangeMessages.size());
        for (int i = 0; i < dataChangeMessages.size(); i++) {
            DataChangeMessage dataChangeMessage = dataChangeMessages.get(i);

            DataMessageBO dataMessageBO = dataMessageConverter.converterBO(dataChangeMessage);

            dataMessageBO.setDiffDataArr(String.join(CoreConstant.COMMA, dataChangeMessage.getUpdateColumns()));
            dataMessageBO.setTableDataJson(JSON.toJSONString(binlogData.getDataMap().get(i)));

            dataMessageBOS.add(dataMessageBO);
        }
        return dataMessageBOS;
    }

    /**
     * 获取变更消息对象
     *
     * @param binlogData
     * @param listenConfigDO
     * @return
     */
    public List<DataChangeMessage> getDataChangeMessage(BinlogData binlogData, DataChangeListenConfigDO listenConfigDO) {

        // 获取监听变更字段配置表信息
        List<DataChangeColumnConfigDO> columnConfigDOS = dataChangeRepository.getColumnConfigByListenId(listenConfigDO.getId());

        // 要监听的字段为空，不作处理
        if (CollectionUtils.isEmpty(columnConfigDOS)) {
            return null;
        }

        // 数据变更消息对象
        return buildChangeColumn(binlogData, columnConfigDOS, listenConfigDO);
    }

    @Override
    public void processBinlogMessage(BinlogData binlogData) {
        // 获取当前表的监听信息
        DataChangeListenConfigDO listenConfigDO = dataChangeRepository.getListenConfigByTable(binlogData.getTableName());

        // 未配置监听信息的表，不作处理
        if (Objects.isNull(listenConfigDO)) {
            return;
        }

        // 数据变更消息对象
        List<DataChangeMessage> dataChangeMessages = getDataChangeMessage(binlogData, listenConfigDO);

        // 不需要监听，或者要监听的字段值未变动
        if (CollectionUtils.isEmpty(dataChangeMessages)) {
            return;
        }

        // 消息模型配置对象
        // 针对这个表数据变更消息, 对他进行再封装发送到 mq
        List<DataChangeMessageConfigBO> messageConfigBOS = dataChangeRepository.getMessageConfigBOByListenId(listenConfigDO.getId());
        // 不需要发送消息
        if (CollectionUtils.isEmpty(messageConfigBOS)) {
            return;
        }

        // 获取需要发送的消息对象
        List<DataSendMessageBO> sendDataMessageList =
                getInternalSendDataMessage(dataChangeMessages, binlogData.getDataMap(), messageConfigBOS);

        // 待发送的消息为空，无需处理
        if (CollectionUtils.isEmpty(sendDataMessageList)) {
            return;
        }

        // 发送消息
        sendDataMessage(sendDataMessageList);

        // 存在外部消息配置的才需要保存外部消息
        if (messageConfigBOS.stream().anyMatch(messageConfigBO ->
                MessageTypeEnum.EXTERNAL_MESSAGE.getCode().equals(messageConfigBO.getMessageType()))) {
            // 保存外部消息详细信息
            saveDataMessageDetail(dataChangeMessages, binlogData);
        }

    }


    /**
     * 发送内部消息
     *
     * @param sendDataMessageList
     */
    private void sendDataMessage(List<DataSendMessageBO> sendDataMessageList) {
        for (DataSendMessageBO dataChangeMessage : sendDataMessageList) {
            DataChangeMessage dataMessage = dataChangeMessage.getDataChangeMessage();
            DataChangeMessageConfigBO dataChangeMessageConfigBO = dataChangeMessage.getDataChangeMessageConfigBO();
            // 发送一个延迟队列的消息出去
            dataMessageProducer.send(dataMessage, dataChangeMessageConfigBO.getMessageTopic(), dataChangeMessageConfigBO.getDelayLevel());
        }
    }

    /**
     * 获取需要发送的消息对象
     *
     * @param dataChangeMessages
     * @param dataMap
     * @param dataChangeMessageConfigBOS
     * @return
     */
    public List<DataSendMessageBO> getInternalSendDataMessage(List<DataChangeMessage> dataChangeMessages,
                                                              List<Map<String, Object>> dataMap,
                                                              List<DataChangeMessageConfigBO> dataChangeMessageConfigBOS) {

        List<DataSendMessageBO> dataSendMessageBOS = new ArrayList<>();
        // 不同系统可以对数据变更消息的监听
        for (DataChangeMessageConfigBO messageConfigBO : dataChangeMessageConfigBOS) {
            // 不是内部消息的不处理
            if (!MessageTypeEnum.INTERNAL_MESSAGE.getCode().equals(messageConfigBO.getMessageType())) {
                continue;
            }
            String notifyColumn = messageConfigBO.getNotifyColumn();
            String[] columns = notifyColumn.split(CoreConstant.COMMA);

            for (int i = 0; i < dataChangeMessages.size(); i++) {
                DataChangeMessage dataChangeMessage = dataChangeMessages.get(i);
                List<DataChangeMessage.ColumnValue> columnValues = new ArrayList<>();
                dataChangeMessage.setColumnValues(columnValues);
                Map<String, Object> data = dataMap.get(i);
                for (String column : columns) {
                    columnValues.add(new DataChangeMessage.ColumnValue(column, data.get(column)));
                }
                dataSendMessageBOS.add(new DataSendMessageBO(messageConfigBO, dataChangeMessage));
            }

        }
        return dataSendMessageBOS;
    }

    /**
     * 构建变更字段
     *
     * @param binlogData
     * @param columnConfigDOS
     * @param listenConfigDO
     * @return
     */
    private List<DataChangeMessage> buildChangeColumn(BinlogData binlogData,
                                                      List<DataChangeColumnConfigDO> columnConfigDOS,
                                                      DataChangeListenConfigDO listenConfigDO) {

        List<DataChangeMessage> dataChangeMessages = new ArrayList<>();

        // 操作类型
        String operateType = binlogData.getOperateType();

        for (int i = 0; i < binlogData.getDataMap().size(); i++) {
            Map<String, Object> data = binlogData.getDataMap().get(i);
            // 新增删除操作
            if (BinlogType.INSERT.getValue().equals(operateType)
                    || BinlogType.DELETE.getValue().equals(operateType)) {
                // 如果是新增，或者删除，oldMap值为null，则所有监听字段都变更
                List<String> updateColumns = columnConfigDOS.stream()
                        .map(DataChangeColumnConfigDO::getListenColumn)
                        .collect(Collectors.toList());
                DataChangeMessage dataChangeMessage =
                        buildDataChangeMessage(binlogData, updateColumns, data.get(listenConfigDO.getKeyColumn()));
                dataChangeMessages.add(dataChangeMessage);
            }
            // 更新操作
            else {
                Map<String, Object> old = binlogData.getOldMap().get(i);

                List<String> updateColumns = new ArrayList<>();
                for (DataChangeColumnConfigDO columnConfigDO : columnConfigDOS) {
                    String column = columnConfigDO.getListenColumn();
                    Object columnOldValue = old.get(column);

                    // 旧的字段值有数据，就表示该字段变更了，添加至修改的字段集合
                    if (!Objects.isNull(columnOldValue)) {
                        updateColumns.add(column);
                    }
                }

                // 监听的字段有数据变更
                if (!CollectionUtils.isEmpty(updateColumns)) {
                    DataChangeMessage dataChangeMessage =
                            buildDataChangeMessage(binlogData, updateColumns, data.get(listenConfigDO.getKeyColumn()));
                    dataChangeMessages.add(dataChangeMessage);
                }
            }
        }

        return dataChangeMessages;
    }

    /**
     * 构建数据变更消息体
     *
     * @param binlogData
     * @param updateColumns
     * @param keyId
     * @return
     */
    private DataChangeMessage buildDataChangeMessage(BinlogData binlogData,
                                                     List<String> updateColumns,
                                                     Object keyId) {
        DataChangeMessage dataChangeMessage =
                new DataChangeMessage(binlogData.getOperateType(), binlogData.getTableName(), updateColumns);
        dataChangeMessage.setMessageNo(SnowflakeIdWorker.getCode());
        dataChangeMessage.setKeyId(keyId);
        return dataChangeMessage;
    }

    /**
     * 处理数据溯源消息
     *
     * @param traceDataList
     */
    @Override
    public void processTraceMessage(List<TraceData> traceDataList) {
        if (CollectionUtils.isEmpty(traceDataList)) {
            return;
        }
        String tableName = traceDataList.get(0).getTableName();
        DataTraceListenConfigDO traceListenConfigByTable = dataChangeRepository.getTraceListenConfigByTable(tableName);
        if (traceListenConfigByTable == null) {
            return;
        }
        dataMessageProducer.send(traceDataList, RocketMqConstant.PRODUCT_TRACE_TOPIC, -1);
    }

    @Override
    public void outDataCacheRefresh() {
        // 1.先获取 缓存中最新的消息消费时间，这里先减去1分钟，避免和 MQ 的处理直接重复
        String createDate = redisCache.get(CacheConstant.CACHE_ROCKET_TIME_KEY);
        if (StrUtil.isEmpty(createDate)) {
            // 如果缓存都不存在最新消费时间，默认处理1个小时内的 DB 数据，再超过前的缓存没有处理必要
            createDate = DateFormatUtil.getHoursDate(-1);
        } else {
            // 缓存时间存在，对时间减去一分钟进行处理
            createDate = DateFormatUtil.getMinuteDate(createDate, -1);
        }

        // 每次处理 DB 兜底的查询之前，先删除掉查询时间范围外的数据，避免数据一直写入导致数据量过大影响性能
        dataRefreshRepository.deleteDataRefresh(DateFormatUtil.getHoursDate(-1));

        //2. 获取是否有超过的最新消费时间的数据落入DB(默认查询超过1分钟还未消费的数据)
        int pageNum = 1;
        // 设置每次查询的数据量，最大为500
        int pageSize = CollectionSize.WRITE_SIZE;
        Page<DataRefreshDetailDO> page = new Page<>(pageNum, pageSize);
        Page<DataRefreshDetailDO> pageResult = dataRefreshRepository.queryDataRefreshDetailDOList(page, createDate);

        List<DataRefreshDetailDO> dataRefreshDetailList = pageResult.getRecords();
        dataRefreshQueue(dataRefreshDetailList);
        try {
            while (pageNum <= pageResult.getPages()) {
                pageNum += 1;
                page.setCurrent(pageNum);
                pageResult = dataRefreshRepository.queryDataRefreshDetailDOList(page, createDate);
                dataRefreshQueue(pageResult.getRecords());
                // 每次循环获取数据后，休眠20ms，避免对数据库造成太大压力
                Thread.sleep(20);
            }
        } catch (InterruptedException e) {
            throw new BaseBizException(ProductExceptionCode.PRODUCT_SQL);
        }
    }

    /**
     * 将数据写入队列进行处理
     *
     * @param dataRefreshDetailList 数据落地的数据
     */
    private void dataRefreshQueue(List<DataRefreshDetailDO> dataRefreshDetailList) {
        if (!CollectionUtils.isEmpty(dataRefreshDetailList)) {
            for (DataRefreshDetailDO dataRefreshDetailDO : dataRefreshDetailList) {
                MessageCache messageCache = dataMessageConverter.converter(dataRefreshDetailDO);
                BlockingQueue blockingQueue = cacheQueue.getBlockingQueue(messageCache.getCacheKey());
                blockingQueue.offer(messageCache);
            }
        }
    }
}
